package local;


import data.handle.*;

import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class LocalInstance {

    private static HandlerThread[] threads;
    private static PartHandle[] partHandles;
    private static LineItemHandle[] lineItemHandles;
    private static ExecutorService executorService;
    public volatile static long runningTime;

    public static void startLocalThreads(int num, int fileReaderThreadNum){

        threads = new HandlerThread[num];
        partHandles = new PartHandle[num];
        lineItemHandles = new LineItemHandle[num];

        int partKeyNum = 2000000 / (num * 4);
        for (int i = 0; i < num; i++) {
            partHandles[i] = new PartHandle(partKeyNum);
            lineItemHandles[i] = new LineItemHandle(partKeyNum, partHandles[i]);
        }

        long beginTimeMillis = System.currentTimeMillis();
        AtomicLong counterFile1 = new AtomicLong(0);
        AtomicLong counterFile2 = new AtomicLong(0);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(num, new Runnable() {

            @Override
            public void run() {
                runningTime = System.currentTimeMillis() - beginTimeMillis;
                System.out.println("use time: " + runningTime + "ms");
                System.out.println("Part all line: " + counterFile1.get());
                System.out.println("LineItem all line: " + counterFile2.get());
            }
        });

        for (int i = 0; i < threads.length; i++) {
            threads[i] = new HandlerThread(fileReaderThreadNum, partHandles[i], lineItemHandles[i]);
            threads[i].setTest(counterFile1, counterFile2, cyclicBarrier);
            new Thread(threads[i]).start();
        }
    }

    public static long getQuantitySum(){
        long sum = 0;
        for (LineItemHandle lineItemHandle : lineItemHandles) {
            sum += lineItemHandle.getQuantitySum();
        }
        return sum;
    }

    public static LinkedBlockingQueue<Result> getResult(byte brandIndex, double quantity){
        LinkedBlockingQueue<Result> results = new LinkedBlockingQueue<>();
        if(executorService == null)
            executorService = Executors.newFixedThreadPool(threads.length);
        CountDownLatch downLatch = new CountDownLatch(threads.length);
        for (LineItemHandle lineItemHandle : lineItemHandles) {
            executorService.execute(new SelectResult(lineItemHandle, brandIndex, quantity, results, downLatch));
        }
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return results;
    }



//    public static void addData(ByteBuffer line, int index, int fileIndex){
//        threads[index].addData(line, fileIndex);
//    }

    public static void addData(PackMessage packMessage, int index, int fileIndex){
        threads[index].addData(packMessage, fileIndex);
    }
}
